package com.concurrent.sample.business.manager.impl;

import cn.hutool.extra.spring.SpringUtil;
import com.concurrent.sample.business.assemble.EventAssemble;
import com.concurrent.sample.business.consts.enums.FlowTypeEnum;
import com.concurrent.sample.business.entity.EventEntity;
import com.concurrent.sample.business.entity.OrderEntity;
import com.concurrent.sample.business.listener.event.CreateTradeEvent;
import com.concurrent.sample.business.listener.event.SysTradeExecReportEvent;
import com.concurrent.sample.business.manager.EventDispatchService;
import com.concurrent.sample.business.consts.BusinessConsts;
import com.concurrent.sample.business.consts.BusinessRedisCacheConsts;
import com.concurrent.sample.business.consts.enums.OrderStatusEnum;
import com.concurrent.sample.business.dto.param.ConfirmTradeEventParam;
import com.concurrent.sample.business.dto.report.TradeExecReportDTO;
import com.concurrent.sample.business.dto.request.CreateTradeEventRequest;
import com.concurrent.sample.business.dto.trade.CreateTradeEventParamParam;
import com.concurrent.sample.business.dto.trade.TradeExecReportParamParam;
import com.concurrent.sample.business.listener.event.ExamineOrderEvent;
import com.concurrent.sample.business.listener.event.ExternalOrderExecReportEvent;
import com.concurrent.sample.business.manager.ITradeEventService;
import com.concurrent.sample.business.service.EventService;
import com.concurrent.sample.business.service.OrderService;
import com.concurrent.sample.common.enums.ExceptionHandleTypeEnum;
import com.concurrent.sample.common.enums.ExpressionTypeEnum;
import com.concurrent.sample.common.lock.DistributedLock;
import com.concurrent.sample.common.lock.DistributedLockTimeOutException;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.interceptor.AsyncExecutionAspectSupport;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.event.EventListener;
import org.springframework.orm.ObjectOptimisticLockingFailureException;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
@RequiredArgsConstructor
@Slf4j
public class EventDispatchServiceImpl implements EventDispatchService {

  private final EventAssemble assemble;

  private final EventService eventService;

  private final OrderService orderService;

  private final ApplicationEventPublisher eventPublisher;

  private final List<ITradeEventService> tradeEventServices;

  /**
   * 监听创建订单请求
   * @param event
   */
  @Async(value = AsyncExecutionAspectSupport.DEFAULT_TASK_EXECUTOR_BEAN_NAME)
  @EventListener(CreateTradeEvent.class)
  public void listenerPlaceTradeEvent(CreateTradeEvent event) {
    SpringUtil.getBean(EventDispatchService.class).placeTrade(event.getPlaceOrderEvents());
  }

  /**
   * 监听审批订单
   * @param examineOrderEvent
   */
  @Async(value = BusinessConsts.ASYNC_EXAMINE_TRADE_EXECUTOR)
  @EventListener(ExamineOrderEvent.class)
  public void listenerTradeExamineEvent(ExamineOrderEvent examineOrderEvent) {
//    SpringUtil.getBean(EventDispatchService.class).autoExamine(examineOrderEvent.getEvents());
    try {
      SpringUtil.getBean(EventDispatchService.class).autoExamine(examineOrderEvent.getEvents());
    } catch (DistributedLockTimeOutException ex) {
      log.warn("do auto examine trade time out exception.");
      // 重新入队处理
      eventPublisher.publishEvent(examineOrderEvent);
    } catch (ObjectOptimisticLockingFailureException ex) {
      log.warn("do auto examine  trade time. optimistic lock. ");
      // 重新入队处理
      eventPublisher.publishEvent(examineOrderEvent);
    }
  }

  /**
   * 监听订单的外部成交回报
   * @param externalOrderExecReportEvent
   */
  @DistributedLock(defaultPrefix = BusinessRedisCacheConsts.TRADE_EXEC_REPORT_LOCK_PREFIX_KEY,
      expression = "#context.execReport.originalOrderId", expressionType = ExpressionTypeEnum.SPEL_EXPRESSION)
  @Async(value = BusinessConsts.ASYNC_TRADE_EXEC_REPORT_EXECUTOR)
  @EventListener(ExternalOrderExecReportEvent.class)
  public void listenerExternalOrderExecReportEvent(ExternalOrderExecReportEvent externalOrderExecReportEvent) {
    TradeExecReportDTO execReport = externalOrderExecReportEvent.getExecReport();

    Long originalOrderId = execReport.getOriginalOrderId();
    OrderEntity order = orderService.getById(originalOrderId);
    if (Objects.isNull(order)) {
      return;
    }
    Long start = System.currentTimeMillis();

    order = orderService.updateByExecutionReport(execReport);
    if(needDoOrderEvent(order, execReport)){
      // 事务完成后调用，注意异步不阻塞主线程
      eventPublisher.publishEvent(new SysTradeExecReportEvent(this, order.getId(), order.getCustomerNo(), execReport));
    }
    log.info("trace order time. handle execution report consuming {} ms.", System.currentTimeMillis() - start);
  }

  /**
   * 监听系统内部的成交回报
   * @param event
   */
  @Async(value = BusinessConsts.ASYNC_TRADE_EXEC_REPORT_EXECUTOR)
  @EventListener(SysTradeExecReportEvent.class)
  public void listenerSysTradeExecReportEvent(SysTradeExecReportEvent event) {
    try {
      SpringUtil.getBean(EventDispatchService.class).handleSysTradeExecReportEvent(event);
    } catch (DistributedLockTimeOutException ex) {
      log.warn("do handle exec report time out exception. orderId = {}, reportId = {}", event.getOrderId(), event.getExecReport().getReportId());
      // 重新入队处理
      eventPublisher.publishEvent(event);
    } catch (ObjectOptimisticLockingFailureException ex) {
      log.warn("do handle exec report. optimistic lock. orderId = {}, reportId = {}", event.getOrderId(), event.getExecReport().getReportId());
      // 重新入队处理
      eventPublisher.publishEvent(event);
    }
  }

  @Override
  public void placeTrade(List<CreateTradeEventRequest> createOrderEventsParams) {
    // 1 创建订单和事件，新启动事务
    List<EventEntity> eventEntities = SpringUtil.getBean(EventDispatchService.class).newTradeEvent(createOrderEventsParams);

    // 2 事务方法外异步自动审批订单，模拟一个个确认订单，此时有并发问题
    for(EventEntity newOrder : eventEntities){
      eventPublisher.publishEvent(new ExamineOrderEvent(this, Collections.singletonList(new ConfirmTradeEventParam(newOrder.getId(), newOrder.getCustomerNo()))));
    }
  }

  @Transactional(rollbackFor = Exception.class)
  @Override
  public List<EventEntity> newTradeEvent(List<CreateTradeEventRequest> createOrderEventsParams) {
    // 先创建事件
    List<EventEntity> eventEntities = createOrderEventsParams.stream()
        .map(param -> assemble.buildEventEntity(param))
        .collect(Collectors.toList());

    eventService.save(eventEntities);

    // 然后创建订单
    List<OrderEntity> orderEntities = eventEntities.stream()
        .map(event -> assemble.buildOrderEntity(event))
        .collect(Collectors.toList());
    orderService.save(orderEntities);

    return eventEntities;
  }

  @Override
  public void autoExamine(List<ConfirmTradeEventParam> confirmTradeEventParams) {
    for(ConfirmTradeEventParam confirmTradeEventParam : confirmTradeEventParams){
      SpringUtil.getBean(EventDispatchService.class).confirm(confirmTradeEventParam);
    }
  }

  @DistributedLock(defaultPrefix = BusinessRedisCacheConsts.CUSTOMER_LOCK_PREFIX_KEY, expression = "#context.customerNo",
      expressionType = ExpressionTypeEnum.SPEL_EXPRESSION)
  @Transactional(rollbackFor = Exception.class)
  @Override
  public void confirm(ConfirmTradeEventParam confirmTradeEventParam) {
    EventEntity eventEntity = eventService.getById(confirmTradeEventParam.getEventId());
    OrderEntity orderEntity = orderService.getByEventId(eventEntity.getId());
    if(FlowTypeEnum.COMMITTED != orderEntity.getFlowStatus()){
      throw new RuntimeException("该" + orderEntity.getFlowStatus() + "订单状态不能被审批确认");
    }
    CreateTradeEventParamParam tradeEventParam = new CreateTradeEventParamParam()
        .setExceptionHandleType(ExceptionHandleTypeEnum.EXCEPTION)
        .setEvent(eventEntity)
        .setOrder(orderEntity);

    ITradeEventService tradeEventService = tradeEventServices.stream()
        .filter(service -> service.support(tradeEventParam)).findFirst().get();

    tradeEventService.handleTrade(tradeEventParam);
  }

  @DistributedLock(defaultPrefix = BusinessRedisCacheConsts.CUSTOMER_LOCK_PREFIX_KEY, expression = "#context.customerNo",
      expressionType = ExpressionTypeEnum.SPEL_EXPRESSION)
  @Transactional(rollbackFor = Exception.class)
  @Override
  public void handleSysTradeExecReportEvent(SysTradeExecReportEvent sysTradeExecReportEvent) {
    OrderEntity orderEntity = orderService.getById(sysTradeExecReportEvent.getOrderId());
    EventEntity eventEntity = eventService.getById(orderEntity.getEventId());

    TradeExecReportParamParam tradeExecReportParam = new TradeExecReportParamParam()
        .setEvent(eventEntity)
        .setOrder(orderEntity)
        .setExecReport(sysTradeExecReportEvent.getExecReport());

    ITradeEventService tradeEventService = tradeEventServices.stream()
        .filter(service -> service.support(tradeExecReportParam)).findFirst().get();

    tradeEventService.handleTradeExecReport(tradeExecReportParam);
  }

  private boolean needDoOrderEvent(OrderEntity order, TradeExecReportDTO report){
    return OrderStatusEnum.isFinished(report.getOrderStatus()) || report.getQuantity() > 0L;
  }

}
